package com.lpssfxy.datainit

import com.lpssfxy.datainit.config.MongoConfig
import com.lpssfxy.datainit.entities.Rating
import com.lpssfxy.datainit.utils.AppUtils
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

object LogParser {

  def main(args: Array[String]): Unit = {
    if (args.length < 1) {
      println("请传入业务系统日志info.log所在路径")
      System.exit(1)
    }
    val logPath = args(0)

    // 创建一个SparkConf配置
    val sparkConf = new SparkConf().setAppName("LogParser") //.setMaster(AppUtils.getSparkCores)
    // 创建一个SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()
    // 在对DataFrame和Dataset进行操作许多操作都需要这个包进行支持
    import spark.implicits._
    // 将Product、Rating数据集加载进来
    val rating2RDD = spark.sparkContext.textFile(logPath) // "/mysoft/jars/recommendersys/businessSys/logs"
    // 进行RDD的转换，过滤出目标日志内容：PRODUCT_RATING_PREFIX:73214636|62138|5.0|1603760681，并处理
    val rating2DF = rating2RDD.filter(_.contains("PRODUCT_RATING_PREFIX:"))
      .map(item => {
        val strings = item.split("PRODUCT_RATING_PREFIX:")(1).trim()
        val arrays = strings.split("\\|")
        Rating(arrays(0).toInt, arrays(1).toInt, arrays(2).toDouble, arrays(3).toInt)
      }).toDF()

    val mongoConfig = MongoConfig(AppUtils.getMongoUri, AppUtils.getMongoDb)
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.uri))

    // 写入rating集合中
    rating2DF.write
      .option("uri", mongoConfig.uri)
      .option("collection", AppUtils.MONGODB_RATING_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()
    // 关闭mongoClient
    mongoClient.close()
    // 停止spark程序
    spark.stop()
  }
}
